package com.wpx.batch;


import com.wpx.batch.exception.WpxBatchRetryException;
import com.wpx.batch.log.ExecuteLogger;
import com.wpx.batch.log.ILogHandler;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class WpxBatchJob<E, F>{

    private ThreadPoolExecutor processorExecutorService;
    private ThreadPoolExecutor writerExecutorService;
    private int maxWaitTask = 8888;  //待处理任务大于此值 将暂停读取，避免读回来的数据过多 占内存
    private int processorThreadPoolSize = 3;
    private int writerThreadPoolSize = 3;
    private ItemReader<E> itemReader;

    private ItemProcessor<E, F> itemProcessor;
    private ItemProcessor<List<E>, List<F>> bacthItemProcessor;
    private int bacthItemProcessorSize = 1;
    private Queue<E> processorDataPool = new ConcurrentLinkedQueue<>();

    private ItemWriter<F> dataWriter;
    private ItemWriter<List<F>> bacthDataWriter;
    private int bacthDataWriterSize = 1;
    private Queue<F> writerDataPool = new ConcurrentLinkedQueue<>();

    private AtomicInteger redaCount = new AtomicInteger(0); //读取数

    private AtomicInteger processorFailCount = new AtomicInteger(0); //处理失败数
    private AtomicInteger processorSuccessCount = new AtomicInteger(0); //已处理数

    private AtomicInteger writerSubCount = new AtomicInteger(0); //已提交到写的

    private AtomicInteger writerFailCount = new AtomicInteger(0); //写失败
    private AtomicInteger writerSuccessCount = new AtomicInteger(0); //已写出数

    private long sleep = 50;
    private boolean isCountRead; //按次数读
    private int readN = 1; //读次数

    private ExecuteContext executeContext;

    private ExecuteLogger.Level logLevel = ExecuteLogger.Level.err;
    private Set<ILogHandler> logHandels = new HashSet<>();

    private Set<Class<? extends Exception>> retryExceptions = new HashSet<>();
    private int retry = 3;

    private boolean willStop;

    private boolean ing = false;

    private Thread assistThread;  //辅助线程，做一些没法在主线程做的辅助功能

    private void init(){
        //统计清理
        redaCount.set(0);
        processorFailCount.set(0);
        processorSuccessCount.set(0);
        writerSubCount.set(0);
        writerFailCount.set(0);
        writerSuccessCount.set(0);
    }

    public void execute(){
        ExecuteParam param = new ExecuteParam();
        execute(param, false);
    }
    public void asyncExecute(){
        ExecuteParam param = new ExecuteParam();
        execute(param, true);
    }
    public void execute(ExecuteParam param){
        execute(param,false);
    }
    public void asyncExecute(ExecuteParam param){
        execute(param,true);
    }

    private void execute(ExecuteParam param, boolean async){
        this.assistThread = new Thread(this::assistStak);
        this.assistThread.start();
        if(async){
            Thread thread = new Thread(() -> {
                star(param);
            });
            thread.setDaemon(false);
            thread.start();
        }else{
            star(param);
        }
    }

    private void star(ExecuteParam param){
        if(ing){
            throw new RuntimeException("不能同时执行!!");
        }
        ing = true;
        willStop = false;
        init();
        long executeId = System.currentTimeMillis();
        executeContext = new ExecuteContext(this, param, new ExecuteLogger(this.logLevel, executeId, this.logHandels));

        int readC = 0;
        List<E> datas = null;
        while (true){
            if(processorExecutorService.getQueue().size() > maxWaitTask){
//                System.out.println("任务量过多暂停读取..");
                WpxBatchUtils.sleep(sleep);  //处理的任务多  休眠一下
                continue;
            }
            //作业将停止  不在读
            if(this.isWillStop()){
                break;
            }

            if(isCountRead){
                if(readC == readN){
                    break;
                }else{
                    readC++;
                }
            }

            datas = itemReader.read(executeContext);
            if(WpxBatchUtils.isEmpty(datas)){
                break;
            }
            this.redaCount.addAndGet(datas.size());

            datas = itemReader.readAfter(datas);
            for (E item : datas) {
                processorDataPool.add(item);
            }
            subProcessor();
        }

        //缓冲处理池 剩余的 提交掉
        flushSubProcessor();



        //等待处理完成
        while(redaCount.get() != (processorSuccessCount.get() + processorFailCount.get())){
            WpxBatchUtils.sleep(500);
//            System.out.println("等待处理出完成");
//            System.out.println("读到的：" + redaCount);
//            System.out.println("在处理的：" + processorExecutorService.getQueue().size());
//            System.out.println("处理cg的：" + processorSuccessCount);
//            System.out.println("处理sb的：" + processorFailCount);
        }
//        processorExecutorService.shutdown();

        //等待已提交的完成
        while(writerSubCount.get() != (writerSuccessCount.get() + writerFailCount.get())){
            WpxBatchUtils.sleep(50);
        }

        //待写池中 剩的的  提交掉
        flushSubWriter();

        //等待写出完成
        while(processorSuccessCount.get() != (writerSuccessCount.get() + writerFailCount.get())){
//            System.out.println("等待写出完成");
//            System.out.println("读到的：" + redaCount);
//            System.out.println("处理cg的：" + processorSuccessCount);
//            System.out.println("处理sb的：" + processorFailCount);
//
//            System.out.println("写出cg的：" + writerSuccessCount);
//            System.out.println("写出sb的：" + writerFailCount);
            WpxBatchUtils.sleep(50);
        }
//        writerExecutorService.shutdown();
        ing = false;
    }

    /**
     * 辅助任务
     */
    private void  assistStak(){
        while(true){
            WpxBatchUtils.sleep(3000);  //3秒跑一次
            //线程任务没有了，尝试提交掉 缓冲的数据，避免缓冲过久
            if(processorExecutorService.getQueue().size() == 0){
                flushSubProcessor();
            }
            if(writerExecutorService.getQueue().size() == 0){
                flushSubWriter();
            }
        }
    }

    /**
     * 将要停止, 关闭读 处理完后 退出
     */
    public void stop(){
        this.willStop = true;
    }

    public boolean isWillStop(){
        return this.willStop;
    }

    /**
     * 是否在运行
     */
    public boolean isIng(){
        return this.ing;
    }


    public static <E, F> WpxBatchJob<E, F> builder(Class<E> eClass, Class<F> fClass){
        WpxBatchJob job = new WpxBatchJob();
        return job;
    }

    public WpxBatchJob logLevel(ExecuteLogger.Level logLevel) {
        this.logLevel = logLevel;
        return this;
    }

    public WpxBatchJob logHandels(ILogHandler... logHandels) {
        if(!WpxBatchUtils.isEmpty(logHandels)){
            for (ILogHandler logHandel : logHandels) {
                this.logHandels.add(logHandel);;
            }
        }
        return this;
    }

    public WpxBatchJob retryExceptions(Class<? extends Exception>... retryExceptionClass) {
        if(!WpxBatchUtils.isEmpty(retryExceptionClass)){
            for (Class<? extends Exception> exceptionClass : retryExceptionClass) {
                this.retryExceptions.add(exceptionClass);
            }
        }
        return this;
    }

    public WpxBatchJob retry(int retry) {
        this.retry = retry;
        return this;
    }

    public WpxBatchJob maxWaitTask(int maxWaitTask){
        this.maxWaitTask = maxWaitTask;
        return this;
    }

    public WpxBatchJob poolSize(int poolSize){
        this.processorThreadPoolSize = poolSize;
        return this;
    }

    public WpxBatchJob processorPoolSize(int poolSize){
        this.processorThreadPoolSize = poolSize;
        return this;
    }
    public WpxBatchJob writerPoolSize(int poolSize){
        this.writerThreadPoolSize = poolSize;
        return this;
    }
    public WpxBatchJob reader(ItemReader<E> dataReader){
        this.itemReader = dataReader;
        return this;
    }
    public WpxBatchJob readerOne(ItemReader<E> dataReader){
        this.itemReader = dataReader;
        this.isCountRead = true;
        this.readN = 1;
        return this;
    }
    public WpxBatchJob reader(int readN, ItemReader<E> dataReader){
        this.itemReader = dataReader;
        this.isCountRead = true;
        this.readN = readN;
        return this;
    }
    public WpxBatchJob processor(ItemProcessor<E, F> itemProcessor){
        this.itemProcessor = itemProcessor;
        return this;
    }
    public WpxBatchJob bacthProcessor(int size, ItemProcessor<List<E>, List<F>> bacthItemProcessor){
        this.bacthItemProcessorSize = size;
        this.bacthItemProcessor = bacthItemProcessor;
        return this;
    }

    public WpxBatchJob writer(ItemWriter<F> dataWriter){
        this.dataWriter = dataWriter;
        return this;
    }

    public WpxBatchJob bacthWriter(int size, ItemWriter<List<F>> bacthDataWriter){
        this.bacthDataWriterSize = size;
        this.bacthDataWriter = bacthDataWriter;
        return this;
    }

    public AtomicInteger getRedaCount() {
        return redaCount;
    }

    public AtomicInteger getProcessorSuccessCount() {
        return processorSuccessCount;
    }

    public AtomicInteger getWriterSuccessCount() {
        return writerSuccessCount;
    }

    public WpxBatchJob build(){
        initExecutorService();
        if(this.itemReader == null){
            throw new RuntimeException("dataReader不为空");
        }
        if(this.itemProcessor == null && this.bacthItemProcessor == null){
            throw new RuntimeException("itemProcessor不为空");
        }
        if(this.itemProcessor != null && this.bacthItemProcessor != null){
            throw new RuntimeException("itemProcessor 单处理和批处理 不能同时存在");
        }
        if(this.dataWriter == null && this.bacthDataWriter == null){
            this.dataWriter = (executeContext, item)->{};
        }
        return this;
    }

    private void initExecutorService(){
        this.processorExecutorService = newFixedThreadPool(this.processorThreadPoolSize);
        this.writerExecutorService = newFixedThreadPool(this.writerThreadPoolSize);
    }

    /**
     * 尝试提交处理
     */
    private void subProcessor(){
        if(itemProcessor != null){
            while(processorDataPool.size() > 0){
                processorExecutorService.submit(new ItemProcessorTask(processorDataPool.poll()));
            }
        }else{
            while(processorDataPool.size() >= bacthItemProcessorSize){
                List<E> list = new ArrayList<>(bacthItemProcessorSize);
                for (int i = 0; i < this.bacthItemProcessorSize; i++) {
                    list.add(processorDataPool.poll());
                }
                processorExecutorService.submit(new ItemBacthProcessorTask(list));
            }
        }
    }

    /**
     * 全部提交处理
     */
    private void flushSubProcessor(){
        if(bacthItemProcessor == null) return;
        List<E> list = new ArrayList<>(bacthItemProcessorSize);
        while(processorDataPool.size() > 0){
            list.add(processorDataPool.poll());
        }
        processorExecutorService.submit(new ItemBacthProcessorTask(list));
    }


    /**
     * 尝试提交写
     */
    private synchronized void subWriter(){
        /**
         * 单写 把待写出池的 清空
         */
        if(dataWriter != null){
            while (writerDataPool.size() > 0){
                writerExecutorService.submit(new ItemWriterTask(writerDataPool.poll()));
                writerSubCount.incrementAndGet();
            }
        }else {
            /**
             * 批量写  一批一批提交写出
             */
            while(writerDataPool.size() >= bacthDataWriterSize){
                List<F> list = new ArrayList<>(bacthDataWriterSize);
                for (int i = 0; i < bacthDataWriterSize; i++) {
                    list.add(writerDataPool.poll());
                }
                writerExecutorService.submit(new ItemBacthWriterTask(list));
                writerSubCount.addAndGet(list.size());
            }
        }

    }

    /**
     * 将待写池中的全部 提交写
     */
    private void flushSubWriter(){
        if(dataWriter != null){
            while (writerDataPool.size() > 0){
                writerExecutorService.submit(new ItemWriterTask(writerDataPool.poll()));
                writerSubCount.incrementAndGet();
            }
        }
        if(bacthDataWriter == null) return;
        List<F> list = new ArrayList<>(writerDataPool.size());
        while (writerDataPool.size() > 0) {
            list.add(writerDataPool.poll());
        }
        writerExecutorService.submit(new ItemBacthWriterTask(list));
        writerSubCount.addAndGet(list.size());
    }


    private class ItemProcessorTask implements Runnable{
        private E item;
        private int retry = WpxBatchJob.this.retry;
        public ItemProcessorTask(E item){
            this.item = item;
        }

        @Override
        public void run() {
            F retuenItem = null;
            try {
                itemProcessor.processorBefter(executeContext, item);
                retuenItem = itemProcessor.processor(executeContext, item);
                itemProcessor.processorAfter(executeContext, item, retuenItem);
            }catch (Exception e){
                if(this.retry > 0){
                    for (Class<? extends Exception> retryException : retryExceptions) {
                        if(retryException == e.getClass()){
                            this.retry--;
                            processorExecutorService.execute(this);
                            return;
                        }
                    }
                }else{
                    e = new WpxBatchRetryException("重试" + WpxBatchJob.this.retry + "次后失败", e);
                }


                processorFailCount.incrementAndGet();
                itemProcessor.processorExceptionHandle(executeContext, e, this.item, retuenItem);
                return;
            }

            processorSuccessCount.incrementAndGet();
            writerDataPool.add(retuenItem);
            subWriter();
        }
    }


    private class ItemBacthProcessorTask implements Runnable{
        private List<E> items;
        private int retry = WpxBatchJob.this.retry;
        public ItemBacthProcessorTask(List<E> items){
            this.items = items;
        }

        @Override
        public void run() {
            List<F> retuenItems = null;
            try {
                bacthItemProcessor.processorBefter(executeContext, items);
                retuenItems = bacthItemProcessor.processor(executeContext, items);
                bacthItemProcessor.processorAfter(executeContext, items, retuenItems);
            }catch (Exception e){
                if(this.retry > 0){
                    for (Class<? extends Exception> retryException : retryExceptions) {
                        if(retryException == e.getClass()){
                            this.retry--;
                            processorExecutorService.execute(this);
                            return;
                        }
                    }
                }else{
                    e = new WpxBatchRetryException("重试" + WpxBatchJob.this.retry + "次后失败", e);
                }

                processorFailCount.addAndGet(items.size());
                bacthItemProcessor.processorExceptionHandle(executeContext, e, this.items, retuenItems);
                return;
            }

            processorSuccessCount.getAndAdd(retuenItems.size());
            for (F retuenItem : retuenItems) {
                writerDataPool.add(retuenItem);
            }

            subWriter();
        }
    }
    private class ItemWriterTask implements Runnable{
        private F item;

        public ItemWriterTask(F item){
            this.item = item;
        }

        @Override
        public void run() {
            try {
                dataWriter.writer(executeContext, this.item);
            }catch (Exception e){
                writerFailCount.incrementAndGet();
                dataWriter.writerExceptionHandle(executeContext, e, item);
                return;
            }

            writerSuccessCount.incrementAndGet();
        }
    }
    private class ItemBacthWriterTask implements Runnable{
        private List<F> items;

        public ItemBacthWriterTask(List<F> items){
            this.items = items;
        }

        @Override
        public void run() {
            try {
                bacthDataWriter.writer(executeContext, this.items);
            }catch (Exception e){
                writerFailCount.addAndGet(this.items.size());
                bacthDataWriter.writerExceptionHandle(executeContext,e, items);
                return;
            }

            writerSuccessCount.getAndAdd(this.items.size());
        }
    }


   private static ThreadPoolExecutor newFixedThreadPool(int nThreads){
       return new ThreadPoolExecutor(nThreads, nThreads,
               0L, TimeUnit.MILLISECONDS,
               new LinkedBlockingQueue<Runnable>());
   }

}
